package shujia;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

public class Demo4Filter {

    Connection conn;
    Admin admin;
    TableName studentTN;
    Table student;

    public void printRSWithFilter(Filter filter) throws IOException {
        Scan scan = new Scan();
        scan.setFilter(filter);

        ResultScanner scanner = student.getScanner(scan);
        for (Result rs : scanner) {
            String id = Bytes.toString(rs.getRow());
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println(id + "," + name + "," + age + "," + gender + "," + clazz);
        }

    }


    @Before
    public void createConn() throws IOException {
        // 1、创建一个配置文件
        Configuration conf = HBaseConfiguration.create();
        // 配置ZK的地址，通过ZK可以找到HBase
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        // 2、创建连接
        conn = ConnectionFactory.createConnection(conf);
        // 3、创建Admin对象
        admin = conn.getAdmin();
        studentTN = TableName.valueOf("student");
        student = conn.getTable(studentTN);

    }

    @Test
    /**
     * age > 23 的学生
     * ValueFilter:作用在每一个cell上，符合要求的cell不会被过滤
     * 结果不符合正常的思维逻辑
     */
    public void ValueFilter1() throws IOException {
        // 创建二进制比较器
        BinaryComparator binaryComparator = new BinaryComparator("23".getBytes());
        // 列值过滤器
        ValueFilter valueFilter = new ValueFilter(CompareFilter.CompareOp.GREATER, binaryComparator);

        printRSWithFilter(valueFilter);

    }

    @Test
    /**
     * SingleColumnValueFilter 单列值过滤器
     * 可以指定一个列进行过滤
     * 该过滤器会将符合过滤条件的列对应的cell所在的整行数据进行返回
     * 如果某条数据的列不符合条件，则会将整条数据进行过滤
     * 如果数据中不存在指定的列，则默认会直接返回
     *
     * age > 23 的学生
     */
    public void SingleColumnValueFilter1() throws IOException {

        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                "info".getBytes(),
                "age".getBytes(),
                CompareFilter.CompareOp.GREATER,
                "23".getBytes()
        );

        printRSWithFilter(singleColumnValueFilter);
    }

    @Test
    /**
     * 过滤文科班的学生
     * SingleColumnValueExcludeFilter 单列值排除过滤器
     * 同单列值过滤器最大的区别在于最后的返回结果包不包含比较的列
     */
    public void SingleColumnValueExcludeFilter2() throws IOException {
        BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("文科".getBytes());

        SingleColumnValueExcludeFilter singleColumnValueExcludeFilter = new SingleColumnValueExcludeFilter(
                "info".getBytes(),
                "clazz".getBytes(),
                CompareFilter.CompareOp.EQUAL,
                binaryPrefixComparator
        );

        printRSWithFilter(singleColumnValueExcludeFilter);

    }

    @Test
    /**
     * 过滤出rowkey（id）以150010088开头的学生
     */
    public void RowFilter1() throws IOException {
        BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("150010088".getBytes());
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);

        printRSWithFilter(rowFilter);

    }

    @Test
    /**
     * 过滤出rowkey（id）以150010088开头的学生
     * 使用PrefixFilter:rowkey前缀过滤器
     * 相当于BinaryPrefixComparator+RowFilter
     */
    public void PrefixFilter1() throws IOException {
        PrefixFilter prefixFilter = new PrefixFilter("150010088".getBytes());

        printRSWithFilter(prefixFilter);
    }

    @Test
    /**
     * 通过正则表达式： [A-Za-z0-9]{1}f[0-9]+ 过滤出符合条件的列簇下的所有cell
     */
    public void RegexFamilyFilter() throws IOException {

        RegexStringComparator regexStringComparator = new RegexStringComparator("[A-Za-z0-9]{1}f[0-9]+");

        FamilyFilter familyFilter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, regexStringComparator);

        Scan scan = new Scan();
        scan.setFilter(familyFilter);

        ResultScanner scanner = student.getScanner(scan);
        for (Result rs : scanner) {
            for (Cell cell : rs.listCells()) {
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String q = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println(cf + ":" + q + " " + value);
            }

        }


    }


    @Test
    /**
     * 过滤出列名包含 q 的所有的cell
     *
     */
    public void SubStringQualifierFilter() throws IOException {

        SubstringComparator comparator = new SubstringComparator("q");

        QualifierFilter qualifierFilter = new QualifierFilter(CompareFilter.CompareOp.EQUAL, comparator);

        Scan scan = new Scan();
        scan.setFilter(qualifierFilter);

        ResultScanner scanner = student.getScanner(scan);
        for (Result rs : scanner) {
            for (Cell cell : rs.listCells()) {
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String q = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println(cf + ":" + q + " " + value);
            }

        }


    }

    @Test
    /**
     * 分页过滤器：PageFilter
     * 获取第四页的数据，每页10条
     *
     * 实际上需要遍历该页前面所有的数据，性能非常低
     */
    public void PageFilter1() throws IOException {
        int page = 4;
        int pageSize = 10;

        // 首先先获取第4页的第一条数据的rk
        int page_first = (page - 1) * pageSize + 1;
        PageFilter pageFilter1 = new PageFilter(page_first);

        Scan scan = new Scan();
        scan.setFilter(pageFilter1);
//        scan.setLimit(40); // PageFilter就相当于setLimit

        String rowkey = null;
        ResultScanner scanner = student.getScanner(scan);
        for (Result rs : scanner) {
            rowkey = Bytes.toString(rs.getRow());
        }

        Scan scan1 = new Scan();
        scan1.withStartRow(rowkey.getBytes());
        PageFilter pageFilter2 = new PageFilter(pageSize);
        scan1.setFilter(pageFilter2);
        ResultScanner scanner2 = student.getScanner(scan1);
        for (Result rs : scanner2) {
            String id = Bytes.toString(rs.getRow());
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println(id + "," + name + "," + age + "," + gender + "," + clazz);
        }

    }

    @Test
    /**
     * 通过合理地设计Rowkey来实现分页的功能
     * 获取第五页的数据 每页还是10条
     *
     */
    public void PageWithRowkey() throws IOException {
        int page = 5;
        int pageSize = 10;

        int baseId = 1500100000;

        int current_page_first_rk = baseId + (page - 1) * pageSize + 1;

        Scan scan = new Scan();
        scan.withStartRow((current_page_first_rk + "").getBytes());
        scan.setLimit(pageSize);

        ResultScanner scanner = student.getScanner(scan);
        for (Result rs : scanner) {
            String id = Bytes.toString(rs.getRow());
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println(id + "," + name + "," + age + "," + gender + "," + clazz);
        }


    }

    @Test
    /**
     * 过滤 gender为男，age>23，理科班的学生
     * 过条件过滤需要使用FilterList
     */
    public void MultipleFilter() throws IOException {

        SingleColumnValueFilter filter1 = new SingleColumnValueFilter(
                "info".getBytes(),
                "gender".getBytes(),
                CompareFilter.CompareOp.EQUAL,
                "男".getBytes()
        );

        SingleColumnValueFilter filter2 = new SingleColumnValueFilter(
                "info".getBytes(),
                "age".getBytes(),
                CompareFilter.CompareOp.GREATER,
                "23".getBytes()
        );

        SingleColumnValueFilter filter3 = new SingleColumnValueFilter(
                "info".getBytes(),
                "clazz".getBytes(),
                CompareFilter.CompareOp.EQUAL,
                new BinaryPrefixComparator("理科".getBytes())
        );
        FilterList filterList = new FilterList();
        filterList.addFilter(filter1);
        filterList.addFilter(filter2);
        filterList.addFilter(filter3);

        printRSWithFilter(filterList);


    }

    @After
    public void close() throws IOException {
        admin.close();
        conn.close();
    }
}
